package com.shujia.flink.source

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction
import org.apache.flink.streaming.api.scala._

object Demo7ParallelSource {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //设置并行度
    env.setParallelism(2)

    val ds: DataStream[Int] = env.addSource(new TestParallelSource)


    ds.print()

    env.execute()

  }
}

/**
  * SourceFunction  单一的sou
  * RichSourceFunction  多了一个open  和close 方法
  * ParallelSourceFunction  并行的source   会根据并行度，启动多个source
  * RichParallelSourceFunction     多了一个open  和close 方法
  */


class TestParallelSource extends ParallelSourceFunction[Int] {


  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {

    val list: List[Int] = List(1, 2, 3, 4, 5)

    for (elem <- list) {
      ctx.collect(elem)
    }

  }

  override def cancel(): Unit = {

  }
}


